package com.spring.distribute.zk.lcok;

import com.spring.distribute.DistributeLock;
import com.spring.distribute.common.Constant;
import com.spring.distribute.zk.ZkLockData;
import com.spring.distribute.zk.service.ZkService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
@Service
public class ZkDistributeLock implements DistributeLock {
    public static final Object lockSync = new Object();

    /**
     *   key: 每个线程自身创建的节点名称
     *
     *   value：
     *          key: threadName  value: 线程名称
     *          key: currentNodeName  value: 当前线程创建的节点名称
     *          key: beforeNodeName   value: 当前线程所需要监听的节点名称
     *          key: monitorStatus    value: 当前线程监听的状态(1: 待监听 2: 成功监听 3: 放弃监听)
     *          key: treeCache        value: 当前线程监听的节点对象
     * */

    @Autowired
    private ZkService zkService;

    private final ThreadLocal<ZkLockData> zkLockDataThreadLocal = new ThreadLocal<ZkLockData>() {
        @Override
        public ZkLockData initialValue() {
            ZkLockData zkLockData = new ZkLockData();
            zkLockData.setThreadName(Thread.currentThread().getName());
            zkLockData.setCurrentNode("");
            zkLockData.setCurrentStatus("");
            zkLockData.setMonitorNode("");
            return zkLockData;
        }
    };

    @Override
    public Boolean lock(String data) {
        Boolean result = Boolean.FALSE;

        String applyResult = applyLock(data);
        if(ZkService.CREATE_FAILED.equals(applyResult)
            || ZkService.INTERRUPT.equals(applyResult)
            || ZkService.CHILDREN_NODE_EMPTY.equals(applyResult)) {

            return Boolean.FALSE;
        }

        if(Boolean.TRUE.toString().equals(applyResult)) {
            return Boolean.TRUE;
        }

        return result;
    }

    private String applyLock(String data) {
        String basePath = ZkService.DISTRIBUTE_LOCK_NODE;
        ZkLockData zkLockData = zkLockDataThreadLocal.get();

        if(Objects.isNull(zkLockData) || StringUtils.isEmpty(zkLockData.getCurrentNode())) {
            String createResult = this.zkService.createNode(basePath + ZkService.SEPARATE, CreateMode.EPHEMERAL_SEQUENTIAL, data);
            if (StringUtils.equals(createResult, ZkService.CREATE_FAILED) || StringUtils.equals(createResult, ZkService.INTERRUPT)) {
                return createResult;
            }

            zkLockData.setCurrentNode(createResult);
            zkLockData.setCurrentStatus(Constant.SUCCESS_CREATE);
            zkLockDataMap.put(zkLockData.getCurrentNode(), zkLockData);
        }

        String currentThreadCreateNode = zkLockData.getCurrentNode();
        List<String> childrenList = this.zkService.getChildrens(basePath);
        if (CollectionUtils.isEmpty(childrenList)) {
            return ZkService.CHILDREN_NODE_EMPTY;
        } else {
            Collections.sort(childrenList);
            if (childrenList.size() <= 1) {
                zkLockData.setCurrentStatus(Constant.LOCKING);
                return Boolean.TRUE.toString();
            }

            String minimumNode = childrenList.get(0);
            if (StringUtils.equals(currentThreadCreateNode, basePath + ZkService.SEPARATE + minimumNode)) {
                zkLockData.setCurrentStatus(Constant.LOCKING);
                return Boolean.TRUE.toString();
            } else {
                int currentNodeIndex = childrenList.indexOf(currentThreadCreateNode.substring(basePath.length() + 1));
                if(currentNodeIndex > 0) {
                    zkLockData.setMonitorNode(basePath + ZkService.SEPARATE + childrenList.get(currentNodeIndex - 1));
                }
            }
        }

        synchronized (ZkDistributeLock.lockSync) {
            Boolean result = waitForLock(zkLockData);
            if (result) {
                return Boolean.TRUE.toString();
            }

            return Boolean.FALSE.toString();
        }
    }

    private Boolean monitorExist(ZkLockData zkLockData) {
        Map<String, ZkLockData> zkLockDataMap = DistributeLock.zkLockDataMap;
        if(zkLockDataMap.containsKey(zkLockData.getMonitorNode())) {
            return Boolean.TRUE;
        }

        return Boolean.FALSE;
    }

    private Boolean waitForLock(ZkLockData zkLockData) {
        Object monitorResult = this.monitorExist(zkLockData);
        if(!((Boolean) monitorResult)) {
            zkLockData.setMonitorNode(null);
            String basePath = ZkService.DISTRIBUTE_LOCK_NODE;
            List<String> childrenList = this.zkService.getChildrens(basePath);
            if (CollectionUtils.isEmpty(childrenList)) {
                return Boolean.FALSE;
            } else {
                Collections.sort(childrenList);
                if (childrenList.size() <= 1) {
                    zkLockData.setCurrentStatus(Constant.LOCKING);
                    return Boolean.TRUE;
                }

                String minimumNode = childrenList.get(0);
                if (StringUtils.equals(zkLockData.getCurrentNode(), basePath + ZkService.SEPARATE + minimumNode)) {
                    zkLockData.setCurrentStatus(Constant.LOCKING);
                    return Boolean.TRUE;
                } else {
                    int currentNodeIndex = childrenList.indexOf(zkLockData.getCurrentNode().substring(basePath.length() + 1));
                    for(int i = currentNodeIndex; i > 0; i--) {
                        String monitorNode = basePath + ZkService.SEPARATE + childrenList.get(i - 1);
                        ZkLockData monitorZkLockData = ZkDistributeLock.zkLockDataMap.get(monitorNode);
                        if(Objects.isNull(monitorZkLockData) || Constant.UN_LOCK.equals(monitorZkLockData.getCurrentStatus())) {
                            return Boolean.TRUE;
                        }

                        if(Constant.LOCKING.endsWith(monitorZkLockData.getCurrentStatus()) || Constant.SUCCESS_CREATE.equals(monitorZkLockData.getCurrentStatus())) {
                            zkLockData.setMonitorNode(monitorNode);
                            break;
                        }

                        // 其他状态则继续
                    }

                    if(zkLockData.getMonitorNode() == null) {
                        return Boolean.FALSE;
                    }
                }
            }
       }

        monitorResult = this.zkService.monitorNode(zkLockData);

        if(monitorResult instanceof Boolean && Boolean.TRUE.equals(monitorResult)) {
            zkLockData.setCurrentStatus(Constant.LOCKING);
            return Boolean.TRUE;
        }

        if(monitorResult instanceof Boolean && Boolean.FALSE.equals(monitorResult)) {
            if(!Objects.isNull(zkLockData.getTreeCacheEvent())) {
                String eventCache = new String(zkLockData.getTreeCacheEvent().getData().getData());
                if (eventCache.equals(TreeCacheEvent.Type.CONNECTION_SUSPENDED.name()) || eventCache.equals(TreeCacheEvent.Type.CONNECTION_LOST.name())) {
                    return Boolean.FALSE;
                }
            }

            this.zkService.deleteNode(zkLockData);
            this.releaseThreadLockData(zkLockData);
            return Boolean.FALSE;
        }

        return Boolean.FALSE;
    }

    @Override
    public Boolean unLock(String data) {
        ZkLockData oldZkLockData = this.zkLockDataThreadLocal.get();
        oldZkLockData.setCurrentStatus(Constant.UN_LOCK);
        oldZkLockData.setMonitorNode(data);
        Boolean unlock = this.zkService.deleteNode(oldZkLockData);
        if(unlock) {
            this.releaseThreadLockData(oldZkLockData);
            return unlock;
        }

        return unlock;
    }

    private void releaseThreadLockData(ZkLockData zkLockData) {
        Boolean delete = DistributeLock.zkLockDataMap.remove(zkLockData.getCurrentNode(), zkLockData);
        // 重置当前线程threadLocal本地数据
        ZkLockData newZkLockData = new ZkLockData();
        newZkLockData.setThreadName(Thread.currentThread().getName());
        newZkLockData.setCurrentNode("");
        newZkLockData.setCurrentStatus("");
        newZkLockData.setMonitorNode("");
        this.zkLockDataThreadLocal.set(newZkLockData);
    }
}
